#include<stdio.h>
#include<queue>
using namespace std;
#include<pthread.h>
#include<unistd.h>

#define THREAD_COUNT 2
//线程安全队列
class RingQueue{
    public:
        RingQueue(){
            capacity = 10;
            pthread_mutex_init(&_que_lock, NULL);
            pthread_cond_init(&_cons_cond, NULL);
            pthread_cond_init(&_prod_cond, NULL);
        }
        ~RingQueue(){
            pthread_mutex_destroy(&_que_lock);
            pthread_cond_destroy(&_cons_cond);
            pthread_cond_destroy(&_prod_cond);
        }
        //提供给生产者线程使用的接口
        void push(int data){
            pthread_mutex_lock(&_que_lock);
            while(_que.size() == capacity){
                pthread_cond_wait(&_prod_cond, &_que_lock);
            }
            _que.push(data);
            printf("I am produce pthread %p: I produce %d\n", pthread_self(), data);
            pthread_mutex_unlock(&_que_lock);
            //通知消费者消费
            pthread_cond_signal(&_cons_cond);
        }
        //提供给消费者线程进行消费的线程
        void pop(int* data){
            pthread_mutex_lock(&_que_lock);
            while(_que.size() == 0){
                pthread_cond_wait(&_cons_cond, &_que_lock);
            }
            *data = _que.front();
            _que.pop();
            printf("I am consume thread %p : I consume %d\n", pthread_self(), *data);
            pthread_mutex_unlock(&_que_lock);
            //通知生产者线程生产
            pthread_cond_signal(&_prod_cond);
        }
    private:
        queue<int> _que;
        size_t capacity;
        //互斥锁
        pthread_mutex_t _que_lock;
        //同步
        pthread_cond_t _cons_cond;
        pthread_cond_t _prod_cond;
};

int g_data = 0;
pthread_mutex_t g_lock = PTHREAD_MUTEX_INITIALIZER;
void* pro_thread_start(void* arg){
    RingQueue* rq = (RingQueue*)arg;
    while(1){
       pthread_mutex_lock(&g_lock);
       rq->push(g_data);
       g_data++;
       //sleep(1);
       pthread_mutex_unlock(&g_lock);
    }
}
void* con_thread_start(void* arg){
    RingQueue* rq = (RingQueue*)arg;
    while(1){
        int data;
        rq->pop(&data);
    }
}
int main(){
    RingQueue* que = new RingQueue();
    pthread_t pro[THREAD_COUNT], con[THREAD_COUNT];
    for(int i = 0; i < THREAD_COUNT; i++){
        int ret = pthread_create(&pro[i], NULL, pro_thread_start, que);
        if(ret < 0){
            perror("pthread_create");
        }
        ret = pthread_create(&con[i], NULL, con_thread_start, que);
        if(ret < 0){
            perror("pthread_create");

        }
    }
    //主线程等待回收工作线程
    for(int i = 0; i < THREAD_COUNT; i++){
        pthread_join(pro[i], NULL);
        pthread_join(con[i], NULL);
    }
    delete que;
    return 0;
}
